package com.diver.flinkdemo.sink;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.diver.flinkdemo.entity.xxt.LastLocation;
import com.diver.flinkdemo.entity.xxt.LocationInfo;
import com.diver.flinkdemo.mapper.dorisMapper.LastLocationMapper;
import com.diver.flinkdemo.mapper.dorisMapper.LocationInfoMapper;
import com.diver.flinkdemo.utils.DynamicDataSourceUtil;
import com.diver.flinkdemo.utils.JedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;

import java.util.Date;

/**
 * @author LuJW
 * @date 2023/4/3 10:56
 */
@Slf4j
public class LastLocationSink extends RichSinkFunction<LastLocation> {
    /**
     * sqlSessionFactory
     */
    private SqlSessionFactory sqlSessionFactory;

    @Override
    public void open(Configuration parameters) throws Exception {

        if (DynamicDataSourceUtil.dorisSessionFactory == null) {
            DynamicDataSourceUtil sourceUtil = new DynamicDataSourceUtil();
            sourceUtil.init();
            JedisUtil.poolInit();
            log.info("二次初始化完成...");
            log.info("jedis二次初始化完成...{}",JedisUtil.getJedis());
        }
        log.info("初始化获取sqlSessionFactory->{}", DynamicDataSourceUtil.dorisSessionFactory);
        sqlSessionFactory = DynamicDataSourceUtil.dorisSessionFactory;

    }

    @Override
    public void invoke(LastLocation value, Context context) throws Exception {
        SqlSession sqlSession = null;
        LastLocationMapper mapper = null ;
        try {
            //获取sqlSession
            log.info("sqlSessionFactory->{}",sqlSessionFactory);
            sqlSession = sqlSessionFactory.openSession();
            mapper = sqlSession.getMapper(LastLocationMapper.class);
            //查询imei是否是第一次入库
            QueryWrapper<LastLocation> wrapper = new QueryWrapper<>();
            wrapper.eq("imei",value.getCardNum());
            if (mapper.selectOne(wrapper)!=null) {
                value.setUpdateTime(new Date());
            }else {
                value.setCreateTime(new Date());
            }
        } catch (Exception e) {
            log.error(this.getClass().getName() + ".invoke()执行失败", e);
            //重试一次
            try {
                if (sqlSession != null) {
                    //查询imei是否是第一次入库
                    QueryWrapper<LastLocation> wrapper = new QueryWrapper<>();
                    wrapper.eq("imei",value.getCardNum());
                    if (mapper.selectOne(wrapper)!=null) {
                        value.setUpdateTime(new Date());
                    }else {
                        value.setCreateTime(new Date());
                    }
                    mapper.insert(value);
                }
            } catch (Exception ex) {
                log.error(this.getClass().getName() + ".invoke()重试执行失败", e);
            }
        } finally {
            //关闭sqlSession
            if (sqlSession != null) {
                log.info("session commit -> session close");
                DynamicDataSourceUtil.sessionDoAfter(sqlSession);

            }
        }
    }
}
